package com.mornd.gulimall.seckill.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.mornd.gulimall.apis.coupon.ICouponClient;
import com.mornd.gulimall.apis.coupon.to.SeckillSessionTo;
import com.mornd.gulimall.apis.coupon.to.SeckillSkuRelationTo;
import com.mornd.gulimall.apis.product.IProductClient;
import com.mornd.gulimall.apis.product.to.SkuInfoTo;
import com.mornd.gulimall.apis.seckill.to.SeckillOrderTo;
import com.mornd.gulimall.common.core.auth.UserHolder;
import com.mornd.gulimall.common.core.exception.ServiceException;
import com.mornd.gulimall.common.core.result.R;
import com.mornd.gulimall.common.core.utils.LocalDateTimeUtil;
import com.mornd.gulimall.seckill.entity.SeckillNowVo;
import com.mornd.gulimall.seckill.service.ISeckillService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.springframework.util.StringUtils.hasText;

/**
 * @author: mornd
 * @dateTime: 2023/8/29 - 9:22
 */

@Slf4j
@Service
public class SeckillServiceImpl implements ISeckillService {
    @Resource
    private ICouponClient couponClient;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private IProductClient productClient;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private RabbitTemplate rabbitTemplate;

    private static final String SESSION_PREFIX = "seckill:sessions:";
    private static final String SKU_PREFIX = "seckill:skus";
    private static final String SKU_STOCK_SEMAPHORE_PREFIX = "seckill:stockSemaphore:"; // 库存信号量
    private static final String USER_PURCHASED_KEY_PREFIX = "seckill:buyed:";

    @Override
    public void upSeckillSku(int day) {
        R<List<SeckillSessionTo>> r = couponClient.latestDaysSeckillSessions(day);
        if (r.isFail()) {
            log.error(r.getMsg());
            throw new ServiceException(r.getMsg());
        }
        List<SeckillSessionTo> data = r.getData();
        if (CollUtil.isEmpty(data)) {
            log.info("秒杀商品为空");
            return;
        }
        saveSessions(data);
        saveSkuInfo(data);
    }

    /**
     * 工具方法
     *
     * @param dateTime
     * @return
     */
    private long toEpochMilli(LocalDateTime dateTime) {
        return LocalDateTimeUtil.toEpochMilli(dateTime);
    }

    private void saveSessions(List<SeckillSessionTo> data) {
        for (SeckillSessionTo session : data) {
            // 转为时间戳，类似于Date的getTime()
            long start = toEpochMilli(session.getStartTime());
            long end = toEpochMilli(session.getEndTime());

            String key = SESSION_PREFIX + start + "-" + end;

            // 每一场秒杀对应哪些商品
            List<String> skuIds = session.getRelationList().stream().map(it -> it.getPromotionSessionId() + "-" + it.getSkuId().toString()).collect(Collectors.toList());
            if (!skuIds.isEmpty()) {
                // set 存储
                stringRedisTemplate.opsForSet()
                        .add(key, skuIds.toArray(new String[skuIds.size()]));

//                stringRedisTemplate.opsForList().leftPushAll(key, skuIds);

                long ttl = end - toEpochMilli(LocalDateTime.now());
                stringRedisTemplate.expire(key, ttl, TimeUnit.MILLISECONDS);
            }
        }
    }

    private void saveSkuInfo(List<SeckillSessionTo> data) {
        BoundHashOperations<String, Object, Object> boundHashOps = stringRedisTemplate.boundHashOps(SKU_PREFIX);
        for (SeckillSessionTo session : data) {
            for (SeckillSkuRelationTo relation : session.getRelationList()) {
                // 场次id + 商品id 保证多个定时任务同一场次不会添加重复的商品
                String uniqueKey = relation.getPromotionSessionId() + "-" + relation.getSkuId();
                if (Boolean.FALSE.equals(boundHashOps.hasKey(uniqueKey))) {
                    // 远程查询 sku 信息
                    R<SkuInfoTo> r = productClient.getSkuInfoById(relation.getSkuId());
                    SkuInfoTo skuInfoTo = r.getData();
                    if (r.isOk() && r.getData() != null) {
                        relation.setSkuInfoTo(skuInfoTo);
                    }

                    long start = toEpochMilli(session.getStartTime());
                    long end = toEpochMilli(session.getEndTime());

                    relation.setSeckillStartTime(start);
                    relation.setSeckillendTime(end);
                    String randomCode = UUID.randomUUID().toString(true);
                    relation.setRandomCode(randomCode);

                    // 设置库存信号量来限流，如果库存有100个，那么就控制100个请求来访问秒杀系统
                    RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE_PREFIX + randomCode);
                    semaphore.trySetPermits(relation.getSeckillCount().intValue());

                    String json = JSON.toJSONString(relation);
                    boundHashOps.put(uniqueKey, json);
                }
            }
        }
    }

    /**
     * 查询当前时间可以秒杀的商品
     *
     * @return
     */
    @Override
    public List<SeckillSkuRelationTo> currentSeckillSkus() {
        List<String> mapResult = new ArrayList<>();

        long now = LocalDateTimeUtil.toEpochMilli(LocalDateTime.now());
        List<String> keys = new ArrayList<>();

        // scan 命令，功能和 keys * 类似，取出符合通配符的key，每次取出count个，直至取完
        long count = 1000;
        String match = SESSION_PREFIX + "*";
        ScanOptions scanOptions = ScanOptions.scanOptions()
                .count(count)
                .match(match).build();

        Cursor<String> cursor = stringRedisTemplate.scan(scanOptions);
        while (cursor.hasNext()) {
            String key = cursor.next();
            // 收集
            keys.add(key);
        }
        cursor.close();

        for (String key : keys) {
            String dateKey = key.replace(SESSION_PREFIX, "");
            String[] split = dateKey.split("-");
            // seckill:sessions:2321312321-32133123123
            long start = Long.parseLong(split[0]);
            long end = Long.parseLong(split[1]);
            if (now >= start && now <= end) {
                // 满足当前日期，取出展示
                Set<String> members = stringRedisTemplate.opsForSet().members(key);
                if (CollUtil.isNotEmpty(members)) {
                    BoundHashOperations<String, String, String> hashOps = stringRedisTemplate.boundHashOps(SKU_PREFIX);
                    // 根据多个key查询 hash
                    List<String> multiGet = hashOps.multiGet(members);
                    if (CollUtil.isNotEmpty(multiGet)) {
                        mapResult.addAll(multiGet);
                    }
                }
            }
        }

        // json 解析为 Java 对象
        List<SeckillSkuRelationTo> result = mapResult.stream().map(it -> {
            SeckillSkuRelationTo relationTo =
                    JSON.parseObject(it, SeckillSkuRelationTo.class);
            return relationTo;
        }).collect(Collectors.toList());

        return result;
    }

    /**
     * 查询某个sku的秒杀信息
     *
     * @param skuId
     * @return
     */
    @Override
    public SeckillSkuRelationTo getSkuSeckillInfo(Long skuId) {
        BoundHashOperations<String, String, String> hashOps = stringRedisTemplate.boundHashOps(SKU_PREFIX);
        Set<String> keys = hashOps.keys();
        if (CollUtil.isNotEmpty(keys)) {
            for (String key : keys) {
                if (key.endsWith("-" + skuId)) {
                    String json = hashOps.get(key);
                    SeckillSkuRelationTo relationTo = JSON.parseObject(json, SeckillSkuRelationTo.class);
                    long now = LocalDateTimeUtil.toEpochMilli(LocalDateTime.now());
                    if (now > relationTo.getSeckillendTime()) {
                        // 已过期，不用处理
                        return null;
                    }
                    if (now < relationTo.getSeckillStartTime()) {
                        // 还未到秒杀时间，隐藏随机码
                        relationTo.setRandomCode(null);
                    }
                    return relationTo;
                }
            }
        }

        return null;
    }

    /**
     * 开始秒杀
     *
     * @param seckillNowVo
     * @return 订单号
     */
    @Override
    public String seckillnow(SeckillNowVo seckillNowVo) {
        long logStartTime = System.currentTimeMillis();
        BoundHashOperations<String, String, String> hashOps = stringRedisTemplate.boundHashOps(SKU_PREFIX);
        String skuKey = seckillNowVo.getSessionId() + "-" + seckillNowVo.getSkuId();
        String value = hashOps.get(skuKey);
        if (!hasText(value)) {
            throw new ServiceException("秒杀商品不存在");
        }

        SeckillSkuRelationTo relationTo = JSON.parseObject(value, SeckillSkuRelationTo.class);
        if (!seckillNowVo.getCode().equals(relationTo.getRandomCode())) {
            throw new ServiceException("秒杀随机码不匹配");
        }
        long now = LocalDateTimeUtil.toEpochMilli(LocalDateTime.now());
        if (now < relationTo.getSeckillStartTime() || now > relationTo.getSeckillendTime()) {
            throw new ServiceException("秒杀尚未开始或秒杀已结束");
        }

        //todo 能购买多少就买多少，然后再给用户提示
        if (seckillNowVo.getNum() > relationTo.getSeckillLimit() || seckillNowVo.getNum() > relationTo.getSeckillCount()) {
            throw new ServiceException("数量已超出");
        }

        // 判断用户是否已经购买过了,或者已经超出购买限制
        String purchasedKey = USER_PURCHASED_KEY_PREFIX + UserHolder.getUser().getId() + ":" + seckillNowVo.getSessionId() + ":" + seckillNowVo.getSkuId();
        String buyCount = stringRedisTemplate.opsForValue().get(purchasedKey);
        // 用户已经抢购的个数
        Integer userBuyedCount = 0;
        if (hasText(buyCount)) {
            // 已经秒杀过，继续判断
            if (userBuyedCount + seckillNowVo.getNum() > relationTo.getSeckillLimit()) {
                throw new ServiceException("每人仅能限购" + relationTo.getSeckillLimit() + "件，您已经抢购了" + userBuyedCount + "件");
            }
            userBuyedCount = Integer.valueOf(buyCount);
        }

        // 该用户未参与该场次下该商品的秒杀
        RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE_PREFIX + seckillNowVo.getCode());

        /**
         * 分布式信号量
         * tryAcquire
         * 查看redis存储的Permits是否等于0，如果等于0则返回false(有等待时间则等待)，否则返回true，并且Permits-当前要获取的个数
         */
        boolean tryAcquire = semaphore.tryAcquire(seckillNowVo.getNum());

        if (tryAcquire) {
            String orderSn = IdWorker.getTimeId();

            // 更新总数量
            relationTo.setSeckillCount(relationTo.getSeckillCount() - seckillNowVo.getNum());
            hashOps.put(skuKey, JSON.toJSONString(relationTo));

            // 更新用户已抢购的数量
            long ttl = relationTo.getSeckillendTime() - now;
            Integer newCount = userBuyedCount + seckillNowVo.getNum();
            stringRedisTemplate.opsForValue().set(purchasedKey, newCount.toString(), ttl, TimeUnit.MILLISECONDS);

            SeckillOrderTo seckillOrderTo = new SeckillOrderTo();
            seckillOrderTo.setOrderSn(orderSn);
            seckillOrderTo.setPromotionSessionId(relationTo.getPromotionSessionId());
            seckillOrderTo.setSkuId(relationTo.getSkuId());
            seckillOrderTo.setNum(seckillNowVo.getNum());
            seckillOrderTo.setSeckillPrice(relationTo.getSeckillPrice());
            seckillOrderTo.setMemberId(UserHolder.getUser().getId());
            seckillOrderTo.setSeckillStartTime(relationTo.getSeckillStartTime());
            seckillOrderTo.setSeckillendTime(relationTo.getSeckillendTime());

            // 发送给mq
            rabbitTemplate.convertAndSend("order-event-exchange",
                    "order.seckill.order",
                    seckillOrderTo);

            long logEndTime = System.currentTimeMillis();

            log.info("秒杀完成，耗时：" + (logEndTime - logStartTime) + "ms"); // 50ms左右
            return orderSn;
        } else {
            throw new ServiceException("未抢购到商品该商品");
        }
    }
}
